package com.moses.cityTraffic.monitor

import java.util.Properties

import com.moses.cityTraffic.utils.{AvgSpeedInfo, TrafficInfo}
import org.apache.flink.api.common.functions.AggregateFunction
import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.api.scala._
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer
import org.apache.flink.util.Collector

object AvgSpeedByMonitorAnalysis {

  def main(args: Array[String]): Unit = {
    // 创建流计算初始化对象
    val streamEnv: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    streamEnv.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

    // 配置kafka
    val props = new Properties()
    props.setProperty("bootstrap.servers", "hadoop102:9092,hadoop103:9092,hadoop104:9092")
    props.setProperty("group.id", "moses_001")

    // 创建一个kafka的source
    // stream1是海量的数据流
    val stream1: DataStream[TrafficInfo] = streamEnv.addSource(new FlinkKafkaConsumer[String]("t_traffic", new SimpleStringSchema(), props).setStartFromEarliest())
      //      val stream1: DataStream[TrafficInfo] = streamEnv.socketTextStream("hadoop102", 9999)
      .map((line: String) => {
        val arr: Array[String] = line.split(",")
        TrafficInfo(arr(0).toLong, arr(1), arr(2), arr(3), arr(4).toDouble, arr(5), arr(6))
        // 引入watermark，并且延时时间为5秒
      }).assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[TrafficInfo](Time.seconds(5)) {
      override def extractTimestamp(t: TrafficInfo) = {
        t.actionTime
      }
    })

    stream1.keyBy(_.monitorId)
      // 时间窗口，窗口大小5分钟，滑动步长1分钟
      .timeWindow(Time.minutes(5), Time.minutes(1))
      .aggregate(
        new AggregateFunction[TrafficInfo, (Double, Long), (Double, Long)] {
          override def createAccumulator() = (0,0)

          override def add(in: TrafficInfo, acc: (Double, Long)) = {
            (acc._1 + in.speed, acc._2 + 1)
          }

          override def getResult(acc: (Double, Long)) = acc;

          override def merge(acc: (Double, Long), acc1: (Double, Long)) = (acc._1 + acc1._1, acc._2 + acc1._2)
        },
        (k:String, w:TimeWindow, input: Iterable[(Double, Long)], out: Collector[AvgSpeedInfo]) => {
          val acc: (Double, Long) = input.last
          val avg: Double = (acc._1 / acc._2).formatted("%.2f").toDouble
          out.collect(AvgSpeedInfo(w.getStart, w.getEnd, k, avg, acc._2.toInt))
        }
      ).addSink()

    streamEnv.execute()
  }



}
